package com.weimob.mcloud.wechatwork.customer.mq;

import com.alibaba.fastjson.JSON;
import com.ctrip.framework.apollo.spring.annotation.ApolloJsonValue;
import com.weimob.artemis.client.consumer.model.ConsumerRecord;
import com.weimob.artemis.client.consumer.model.MessageResult;
import com.weimob.artemis.client.spring.annotation.ArtemisConsumerListener;
import com.weimob.artemis.client.spring.annotation.Subscribe;
import com.weimob.mcloud.wechatwork.customer.config.EmployeeStatusChangeConfig;
import com.weimob.mcloud.wechatwork.customer.domain.ArtemisConstant;
import com.weimob.mcloud.wechatwork.customer.domain.EmployeeAuthorizeChangeInfo;
import com.weimob.mcloud.wechatwork.customer.mq.handler.employee.EmployeeAuthorizeChangeHandler;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.CompletableFuture;

/**
 * 员工授权范围变更消息消费
 */
@Component
@Slf4j
public class EmployeeAuthorizeChangeConsumer {

    @Autowired
    @Qualifier("employeeAuthorizeChangeThreadPool")
    private ThreadPoolTaskExecutor employeeAuthorizeChangeThreadPool;

    @Autowired
    @Qualifier("employeeSeatChangeThreadPool")
    private ThreadPoolTaskExecutor employeeSeatChangeThreadPool;

    @Autowired
    private EmployeeAuthorizeChangeHandler employeeAuthorizeChangeHandler;

    @ApolloJsonValue("${employee.status.change.config:{}}")
    private EmployeeStatusChangeConfig employeeStatusChangeConfig;

    /**
     * 员工授权范围变更消息消费
     */
    @ArtemisConsumerListener(
            value = @Subscribe(topic = ArtemisConstant.Topic.WECHATWORK_EMPLOYEE_DEP_AUTHORIZE_MSG, user = ArtemisConstant.CONSUMER_USER), pollSize = 5)
    public void employeeAuthorizeChangeConsumer(MessageResult result) {
        List<ConsumerRecord> records = result.getRecords();
        CompletableFuture<Void>[] completableFutures = new CompletableFuture[records.size()];

        for (int i = 0; i < records.size(); i++) {
            ConsumerRecord consumerRecord = records.get(i);
            completableFutures[i] = CompletableFuture.runAsync(() -> {
                long itemStartTime = System.currentTimeMillis();
                try {
                    EmployeeAuthorizeChangeInfo employeeAuthorizeChangeInfo = JSON.parseObject(consumerRecord.getValue(), EmployeeAuthorizeChangeInfo.class);
                    if (CollectionUtils.isNotEmpty(employeeStatusChangeConfig.getEmployeeAuthChangeCorpidBlackList())
                            && employeeStatusChangeConfig.getEmployeeAuthChangeCorpidBlackList().contains(employeeAuthorizeChangeInfo.getCorpId())) {
                        log.info("黑名单拦截客户消费员工授权范围变更 message：{}", JSON.toJSONString(employeeAuthorizeChangeInfo));
                        return;
                    }
                    log.info("员工授权范围变更消息消费,原始消息：{}", employeeAuthorizeChangeInfo);
                    //valid
                    List<String> employeeIds = employeeAuthorizeChangeInfo.getEmployeeIds();
                    String changeType = employeeAuthorizeChangeInfo.getChangeType();
                    String corpId = employeeAuthorizeChangeInfo.getCorpId();
                    if (CollectionUtils.isEmpty(employeeIds) || StringUtils.isBlank(changeType) || StringUtils.isBlank(corpId)) {
                        log.error("employeeAuthorizeChangeConsumer valid failed, param is null");
                        return;
                    }
                    employeeAuthorizeChangeHandler.handleEmployeeAuthorizeChange(employeeAuthorizeChangeInfo);
                    log.info("客户消费员工授权范围变更结束,employeeIds：{}, cost: {}", JSON.toJSON(employeeIds), System.currentTimeMillis() - itemStartTime);
                } catch (Exception e) {
                    log.error("客户消费员工授权范围变更 Artemis数据处理失败. error", e);
                }

            }, employeeAuthorizeChangeThreadPool);

        }
        //等待执行完成再提交offset
        CompletableFuture.allOf(completableFutures).join();
    }


    /**
     * 员工坐席限制变更消息消费
     */
    @ArtemisConsumerListener(
            value = @Subscribe(topic = ArtemisConstant.Topic.WECHATWORK_EMPLOYEE_SEAT_AUTH_MSG, user = ArtemisConstant.CONSUMER_USER), pollSize = 5)
    public void employeeSeatAuthChangeConsumer(MessageResult result) {
        List<ConsumerRecord> records = result.getRecords();
        CompletableFuture<Void>[] completableFutures = new CompletableFuture[records.size()];

        for (int i = 0; i < records.size(); i++) {

            ConsumerRecord consumerRecord = records.get(i);
            completableFutures[i] = CompletableFuture.runAsync(() -> {
                long itemStartTime = System.currentTimeMillis();
                try {
                    EmployeeAuthorizeChangeInfo employeeAuthorizeChangeInfo = JSON.parseObject(consumerRecord.getValue(), EmployeeAuthorizeChangeInfo.class);
                    if (CollectionUtils.isNotEmpty(employeeStatusChangeConfig.getEmployeeSeatStatusChangeCorpidBlackList())
                            && employeeStatusChangeConfig.getEmployeeSeatStatusChangeCorpidBlackList().contains(employeeAuthorizeChangeInfo.getCorpId())) {
                        log.info("黑名单拦截消费员工授权范围变更 message：{}", JSON.toJSONString(employeeAuthorizeChangeInfo));
                        return;
                    }
                    log.info("员工坐席变更消息消费,原始消息：{}", employeeAuthorizeChangeInfo);
                    //valid
                    List<String> employeeIds = employeeAuthorizeChangeInfo.getEmployeeIds();
                    String changeType = employeeAuthorizeChangeInfo.getChangeType();
                    String corpId = employeeAuthorizeChangeInfo.getCorpId();
                    if (CollectionUtils.isEmpty(employeeIds) || StringUtils.isBlank(changeType) || StringUtils.isBlank(corpId)) {
                        log.error("employeeSeatAuthChangeConsumer valid failed, param is null");
                        return;
                    }
                    employeeAuthorizeChangeHandler.handleEmployeeAuthorizeChange(employeeAuthorizeChangeInfo);
                    log.info("客户消费员工坐席变更结束,employeeIds:{}, cost: {}", JSON.toJSON(employeeIds), System.currentTimeMillis() - itemStartTime);
                } catch (Exception e) {
                    log.error("客户消费员工坐席变更 Artemis数据处理失败. error", e);
                }
            }, employeeSeatChangeThreadPool);

        }
        //等待执行完成再提交offset
        CompletableFuture.allOf(completableFutures).join();
    }

}
